package com.ruoyi.flink.util.read;

import java.util.Map;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;

import com.alibaba.fastjson.JSONObject;
import com.ruoyi.flink.util.db.ClickHouseLink;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

public class RedisRead extends RichSourceFunction<Map<String, String>>{

	private static final long serialVersionUID = 1L;
	String key;
	Jedis jedis = null;
	JedisPool jedisPool = null;
	
	public RedisRead(String key){
		this.key = key;
	}

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        JedisPoolConfig config = new JedisPoolConfig();
		config.setMaxTotal(1000);//jedis的最大活跃连接数
		config.setMaxIdle(50);//jedis最大空闲连接数
		config.setMaxWaitMillis(-1);//jedis池没有连接对象返回时，等待可用连接的最大时间，单位毫秒，默认值为-1，表示永不超时。 如果超过等待时间，则直接抛出JedisConnectionException
        config.setTestOnBorrow(true);//从池中获取连接的时候，是否进行有效检查
		config.setTestOnReturn(true);//归还连接的时候，是否进行有效检查
        jedisPool = new JedisPool(config, "127.0.0.1", 6379, 0, "rick");
           
    }
	@Override
	public void run(SourceContext<Map<String, String>> ctx) throws Exception {	
		ctx.collect(jedisPool.getResource().hgetAll(key));		
	}

	@Override
	public void cancel() {
		if (jedis != null) {
			jedis.close();
		}
		if (jedisPool != null) {
			jedisPool.close();
		}
		
	}

}
